package com.gin.table;

import com.gin.table.util.ClasspathSqlLoader;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.IOException;
import java.util.List;

/**
 * 每30秒统计一次
 * 消费kafka消息, 使用 json format处理复杂json对象
 *
 * 生产者:
 * kafka-console-producer.sh --broker-list dev--194-111:9092  --topic kafka-json-sql-test-topic
 *
 * 消息:
 * {"funcName":"test","data":{"snapshots":[{"content_type":"application/x-gzip-compressed-jpeg","url":"https://blog.csdn.net/xianpanjia4616"}],"audio":[{"content_type":"audio/wav","url":"https://blog.csdn.net"}]},"resultMap":{"result":{"cover":"/data/test/log.txt"},"isSuccess":true},"meta":{"video_type":"normal"},"type":2,"timestamp":1610550007263,"arr":[{"address":"北京市海淀区1","city":"bj"},{"address":"北京市海淀区2","city":"bj"},{"address":"北京市海淀区3","city":"bj"}],"map":{"flink":456},"doublemap":{"inner_map":{"key":123}}}
 *
 * @author gin
 * @date 2021/4/22
 */
public class TestJsonSqlWindow {

    public static void main(String[] args) throws IOException {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        List<String> sqlList = ClasspathSqlLoader.getSqlList("sql/kafka-json.sql");

        TableResult tableResult = null;
        if (!sqlList.isEmpty()) {
            for (String sql : sqlList) {
                System.out.println("executing sql: " + sql);
                tableResult = tEnv.executeSql(sql);
            }
        }
        if (tableResult != null) {
            tableResult.print();
        } else {
            System.out.println("null");
        }

    }

}
